package cn.itcast.flink.transformation;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lilulu
 */

/**
 * flink流中的max、min、maxBy,minBy算子
 */
public class TransformationMaxMinDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Tuple3<String, String, Integer>> dataStreamSource = env.fromElements(
                Tuple3.of("上海", "浦东新区", 777),
                Tuple3.of("上海", "闵行区", 999),
                Tuple3.of("上海", "杨浦区", 666),
                Tuple3.of("北京", "东城区", 567),
                Tuple3.of("北京", "西城区", 987),
                Tuple3.of("上海", "静安区", 888),
                Tuple3.of("北京", "海淀区", 9999)
        );
        // 3. 数据转换-transformation
        // todo: max 最大值, 只关心指定字段最大值，其他字段不关心
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> max = dataStreamSource.keyBy(tuple -> tuple.f0).max(2);
//        max.print("max>");
        // todo: maxBy 最大值，关心其他字段
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> maxBy = dataStreamSource.keyBy(tuple -> tuple.f0).maxBy(2);
//        maxBy.print("maxBy>");
        // todo: max 最小值, 只关心指定字段最小值，其他字段不关心
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> min = dataStreamSource.keyBy(tuple -> tuple.f0).min(2);
//        min.print("min>");
        // todo: maxBy 最小值，关心其他字段
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> minBy = dataStreamSource.keyBy(tuple -> tuple.f0).minBy(2);
        minBy.print("minBy>");
        // 4. 数据终端-sink
        // 5. 触发执行-execute
        env.execute("TransformationMaxMinDemo");
    }
}